package org.com.blbl.ViewCounts;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
import java.util.Map;
import java.util.TreeMap;

public class ViewCountsReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    private TreeMap<Long, String> sortedMap = new TreeMap<>((a, b) -> b.compareTo(a)); // 倒序排序

    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        long sum = 0;
        for (LongWritable value : values) {
            sum += value.get();
        }
        // Add the sum and key to the TreeMap for sorting
        sortedMap.put(sum, key.toString());
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // Output sorted results
        for (Map.Entry<Long, String> entry : sortedMap.entrySet()) {
            context.write(new Text(entry.getValue()), new LongWritable(entry.getKey()));
        }
    }
}
